/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.service.fragmentation;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Hashtable;

import cgl.narada.event.EventHeaders;
import cgl.narada.event.NBEvent;
import cgl.narada.event.impl.NBEventGenerator;
import cgl.narada.service.ServiceBulletin;
import cgl.narada.service.ServiceBulletinManager;
import cgl.narada.service.ServiceException;
import cgl.narada.service.ServiceNotification;
import cgl.narada.service.ServiceProperties;
import cgl.narada.service.Task;
import cgl.narada.service.TaskManager;
import cgl.narada.service.client.EventConsumer;
import cgl.narada.service.qos.impl.QosServiceImpl;

/**
 * This class is responsible for coalescing fragmented events.
 * 
 * @author Shrideep Pallickara $Date$ $Revision$
 */

public class CoalescingService {

    private QosServiceImpl qosServiceImpl;

    private NBEventGenerator nbEventGenerator;

    private String defragDirectory = "C:/TempFiles/tmpFiles/coalesce";

    private String fragExt = ".nbf";

    private String epochExt = ".nbepk";

    private String moduleName = "CoalescingService: ";

    /** fragmentId -> FragmentArray */
    private Hashtable fusionTable;

    /** fragmentId -> consumer */
    private Hashtable consumerTable;

    /** fragmentId -> coalescingMetadata */
    private Hashtable metadataTable;

    private Hashtable taskTable;

    private Hashtable epochs;

    private boolean debug = false;

    protected CoalescingService() {
        fusionTable = new Hashtable();
        consumerTable = new Hashtable();
        metadataTable = new Hashtable();
        taskTable = new Hashtable();
        epochs = new Hashtable();
        this.nbEventGenerator = new NBEventGenerator();
    }

    public CoalescingService(QosServiceImpl qosServiceImpl) {
        this.qosServiceImpl = qosServiceImpl;
        ServiceProperties serviceProperties = qosServiceImpl
            .getServiceProperties();
        String dirValue = serviceProperties.getProperty("CoalescingDirectory");
        if (dirValue != null) {
            defragDirectory = dirValue;
            if (debug) {
                System.out.println(moduleName + "Coalescing Directory value = "
                    + defragDirectory);
            }
        }

        this.nbEventGenerator = qosServiceImpl.getEventGenerator();
        fusionTable = new Hashtable();
        consumerTable = new Hashtable();
        metadataTable = new Hashtable();
        taskTable = new Hashtable();
        epochs = new Hashtable();
        populateEpochs();
    }

    private void populateEpochs() {
        File dir = new File(defragDirectory);

        if (!dir.exists()) {
            return;
        }

        if (!dir.isDirectory()) {
            return;
        }

        String[] fileList = dir.list();
        if (fileList == null) {
            return;
        }

        for (int i = 0; i < fileList.length; i++) {
            if (fileList[i].endsWith(epochExt)) {
                String toAdd = fileList[i].substring(0, fileList[i].length()
                    - epochExt.length());
                epochs.put(toAdd, toAdd);
            }
        }

    }

    public synchronized void manageFragmentedEvent(EventConsumer consumer,
                                                   NBEvent fragmentedEvent)
                                                                           throws ServiceException {
        if (!fragmentedEvent.hasEventHeaders()) {
            System.out.println(moduleName + "Event does not contain any event "
                + "headers. Event will not be processed further.");
            return;
        }

        String fragmentationId = fragmentedEvent.getEventHeaders()
            .getFragmentationIdentifier();
        if (fragmentationId == null) {
            System.out.println(moduleName
                + "FragmentationId == null in the event. "
                + "Event will not be processed further.");
            return;
        }

        int fragmentNumber = fragmentedEvent.getEventHeaders()
            .getFragmentNumber();
        String epochKey = fragmentedEvent.getEventHeaders()
            .getFragmentationEpoch()
            + "";
        if (epochs.containsKey(epochKey)) {
            System.out.println(moduleName + "Fragment (" + fragmentNumber
                + ") of file=" + fragmentationId + " was previously "
                + "processed!");
            return;
        }

        CoalescingMetaData cmd = getCoalescingTableEntry(consumer,
                                                         fragmentedEvent);

        String coalescingDir = cmd.getCoalescingDirectory();
        byte[] fragmentedPayload = fragmentedEvent.getContentPayload();

        System.out.println("\n\n");
        System.out.println(moduleName + "Processing fragment ("
            + fragmentNumber + ") of file=" + fragmentationId);

        writeIndividualFragment(coalescingDir, fragmentNumber,
                                fragmentedPayload);

        reflectTaskProgression(fragmentationId, 0,
                               getNumOfFragmentsAvailable(coalescingDir));

        if (checkIfAllFragmentsAvailable(cmd)) {
            prepareAndReleaseCoalescedEvent(cmd, fragmentedEvent);
        }

        /** Control memory utilization */
        fragmentedEvent = null;
        long totalMemory = Runtime.getRuntime().totalMemory();
        System.out.println(moduleName + "Total Memory Utilization is  ["
            + totalMemory + "] bytes");
    }

    private int getNumOfFragmentsAvailable(String coalescingDir) {
        File coalescingDirectory = new File(coalescingDir);

        if (!coalescingDirectory.exists()) {
            return 0;
        }

        if (!coalescingDirectory.isDirectory()) {
            return 0;
        }

        String[] files = coalescingDirectory.list();
        if (files == null) {
            return 0;
        }

        return files.length;
    }

    private void prepareAndReleaseCoalescedEvent(CoalescingMetaData cmd,
                                                 NBEvent fragmentedEvent)
                                                                         throws ServiceException {
        String filename = cmd.getFileName();
        String coalescedFileLocation = createCoalescedFile(cmd);
        System.out.println(moduleName + "Created coalesced file ->"
            + coalescedFileLocation);

        byte[] coalescedPayload = coalescedFileLocation.getBytes();

        NBEvent coalescedEvent = nbEventGenerator
            .getCoalescedEvent(fragmentedEvent, coalescedPayload);

        EventConsumer consumer = (EventConsumer) consumerTable.get(filename);
        consumer.getNBEventListener().onEvent(coalescedEvent);

        /**
         * Release entries corresponding to the fragmentationId in the fusion
         * and consumer tables
         */
        fusionTable.remove(filename);
        consumerTable.remove(filename);
        metadataTable.remove(filename);
    }

    private CoalescingMetaData getCoalescingTableEntry(EventConsumer consumer,
                                                       NBEvent fragmentedEvent) {

        EventHeaders headers = fragmentedEvent.getEventHeaders();
        String filename = headers.getFragmentationIdentifier();

        CoalescingMetaData cmd;
        if (metadataTable.containsKey(filename)) {
            cmd = (CoalescingMetaData) metadataTable.get(filename);
            return cmd;
        }

        int totalNumOfFragments = headers.getTotalNumberOfFragments();
        long contentLength = headers.getOriginalPayloadLength();
        long fragmentationEpoch = headers.getFragmentationEpoch();
        String coalescingDir = defragDirectory + "/" + filename + "-tempDir";

        cmd = new CoalescingMetaData(filename, coalescingDir, contentLength,
                                     totalNumOfFragments, fragmentationEpoch);
        metadataTable.put(filename, cmd);
        consumerTable.put(filename, consumer);
        initializeTask(filename, totalNumOfFragments);

        return cmd;
    }

    /**
     * This method is used to write an individual fragment to the specified
     * location.
     */
    private void writeIndividualFragment(String fragDirLocation,
                                         int fragmentNum, byte[] fragment)
                                                                          throws ServiceException {
        try {

            /**
             * Make sure that the temporary fragmentation directory exists. If
             * it does not proceed to create the directory
             */
            File fragDirectory = new File(fragDirLocation);
            if (!fragDirectory.exists()) {
                fragDirectory.mkdirs();
                System.out.println(moduleName + "Created directory "
                    + fragDirLocation);
            }

            /**
             * Create the location of the fragmentation file that we would be
             * writing the fragment to
             */
            String fragFileName = fragDirLocation + "/fragment" + fragmentNum
                + fragExt;
            File fragFile = new File(fragFileName);
            if (!fragFile.createNewFile()) {
                System.out.println(moduleName + "fragment [" + fragFileName
                    + "] exists. Will re-write!");
            }

            FileOutputStream fos = new FileOutputStream(fragFile);
            fos.write(fragment);
            fos.close();

            System.out.println(moduleName + "Wrote fragment ["
                + fragFile.getName() + "] of size = (" + fragment.length + ")");
            if (fragFile.length() == 0) {
                System.out.println(moduleName + "fragment ["
                    + fragFile.getName() + " does not exist. \n\n\n\n");
            }

        } catch (IOException ioe) {
            throw new ServiceException(moduleName
                + "Error in writeIndividualFragment()->" + ioe);
        }
    }

    private boolean checkIfAllFragmentsAvailable(CoalescingMetaData cmd)
                                                                        throws ServiceException {

        String filename = cmd.getFileName();
        int totalNumOfFragments = cmd.getNumberOfFragments();
        long contentLength = cmd.getContentLength();
        String dirName = cmd.getCoalescingDirectory();

        File fragDirectory = new File(dirName);
        File[] _fragmentFiles = fragDirectory.listFiles();

        /*
         * System.out.println(moduleName + "Listing fragments available in " +
         * "directory " + dirName ); for (int j=0; j < _fragmentFiles.length;
         * j++) { System.out.println(moduleName + "[" +
         * _fragmentFiles[j].getName() + ", size=" + _fragmentFiles[j].length() ); }
         */

        long totalFileSize = 0;
        for (int i = 1; i <= totalNumOfFragments; i++) {
            String fragFileName = dirName + "/fragment" + i + ".nbf";
            File fragFile = new File(fragFileName);
            // System.out.print(" " + i);
            // if (!fragFile.isFile()) {
            if (fragFile.length() == 0) {
                System.out.println(moduleName + "Fragment [" + i + "] =>"
                    + fragFileName + " Of (" + totalNumOfFragments
                    + ") TOTAL Fragments has not been received yet");
                return false;
            }
            totalFileSize += fragFile.length();
        }/* end for */

        if (totalFileSize != contentLength) {
            throw new ServiceException(moduleName
                + "There is a disparity in the sizes "
                + " of fragments for file [" + filename + "], total "
                + "size of accumulated fragments=" + totalFileSize
                + ", while the original content length=" + contentLength
                + "\n\n\n");
        }

        System.out.println(moduleName
            + "Fragments are available for coalescing"
            + " The coalesced file is of size = " + totalFileSize);
        return true;
    }

    private String createCoalescedFile(CoalescingMetaData cmd)
                                                              throws ServiceException {

        String filename = cmd.getFileName();
        int totalNumOfFragments = cmd.getNumberOfFragments();
        long contentLength = cmd.getContentLength();
        String dirName = cmd.getCoalescingDirectory();

        File dir = new File(dirName);
        String coalescedFileLocation = dir.getParent() + "/" + filename;
        try {
            File coalescedFile = new File(coalescedFileLocation);
            coalescedFile.createNewFile();
            FileOutputStream fos = new FileOutputStream(coalescedFile);

            for (int i = 1; i <= totalNumOfFragments; i++) {
                String fragFileName = dirName + "//fragment" + i + fragExt;

                File fragFile = new File(fragFileName);
                FileInputStream fis = new FileInputStream(fragFile);
                int willRead = (int) fragFile.length();
                byte[] fragment = new byte[willRead];
                int bytesActuallyRead = fis.read(fragment);
                if (willRead != bytesActuallyRead) {
                    System.out.println(moduleName + "Tried to read ("
                        + willRead + ") bytes. Ended up reading only ("
                        + bytesActuallyRead + ") bytes");
                }

                fis.close();

                fos.write(fragment);
                reflectTaskProgression(filename, 1, i);
            } /* end totalNumberOfFragments */
            fos.close();

            System.out.println("\n\n\n\n" + moduleName + "Will delete ("
                + totalNumOfFragments + ") fragments");

            for (int i = 1; i <= totalNumOfFragments; i++) {
                String fragFileName = dirName + "//fragment" + i + fragExt;
                File fragFile = new File(fragFileName);
                System.out.println(moduleName + "Deleting ["
                    + fragFile.getName() + "]");
                fragFile.delete();
                reflectTaskProgression(filename, 2, i);
            }

            String epochFileLoc = dir.getParent() + "/"
                + cmd.getFragmentationEpoch() + epochExt;
            File epochFile = new File(epochFileLoc);
            epochFile.createNewFile();
            String epochKey = cmd.getFragmentationEpoch() + "";
            epochs.put(epochKey, epochKey);

            System.out.println("\n\n\n\n");

        } catch (IOException ioe) {
            throw new ServiceException(moduleName
                + "Error in createCoalescedFile()" + "-> [" + filename + "] "
                + ioe);
        }

        return coalescedFileLocation;
    }

    private void initializeTask(String filename, int numOfFragments) {
        if (taskTable.containsKey(filename)) {
            System.out.println(moduleName
                + "Removing task info pertaining to a"
                + " previous file transfer");
            taskTable.remove(filename);
        }

        Task task = TaskManager.createTask();
        TaskManager.setTaskDescription(task, "The robust receipt of file ["
            + filename + "]");
        TaskManager.createSubTasks(task, 3);

        Task subTask0 = task.getSubTask(0);
        TaskManager.setTaskDescription(subTask0, "Writing file fragments");
        TaskManager.setNumOfWorkUnits(subTask0, numOfFragments);

        Task subTask1 = task.getSubTask(1);
        TaskManager.setTaskDescription(subTask1, "Coalescing file fragments");
        TaskManager.setNumOfWorkUnits(subTask1, numOfFragments);

        Task subTask2 = task.getSubTask(2);
        TaskManager.setTaskDescription(subTask2, "Deleting file fragments");
        TaskManager.setNumOfWorkUnits(subTask2, numOfFragments);

        taskTable.put(filename, task);
    }

    private void reflectTaskProgression(String filename, int subTaskNum,
                                        int numOfWorkUnitsCompleted) {
        Task task = (Task) taskTable.get(filename);

        if (task == null) {
            System.out.println(moduleName
                + "Trying to update task progression on" + " an unknown task!");
            return;
        }

        Task subTask = task.getSubTask(subTaskNum);
        TaskManager
            .setNumOfWorkUnitsCompleted(subTask, numOfWorkUnitsCompleted);

        if (!metadataTable.containsKey(filename)) {
            System.out.println(moduleName
                + "Metadata not available for filename " + filename);
            return;
        }

        int entityId = qosServiceImpl.getEntityId();
        issueServiceNotification(entityId, task);
    }

    private void issueServiceNotification(int entityId, Task taskUpdate) {
        ServiceBulletin bulletin = ServiceBulletinManager
            .getServiceBulletin(entityId);

        if (bulletin == null) {
            System.out.println(moduleName
                + "No ServiceBulletin registered for " + "entity [" + entityId
                + "]");
            return;
        }

        ServiceNotification notification = new ServiceNotification();
        notification.addTaskInfo(taskUpdate);
        bulletin.onServiceNotification(notification);
    }

}
